package com.example.adminservice.kafka;

import com.example.commom.enums.StatusCodeEnum;
import com.example.commom.exception.CustomException;
import com.example.commom.global.KafkaConstants;
import com.example.commom.redis.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;


/**
 *  专门用来处理用户登录状态的 kafka 消费者
 */
@Slf4j
@Component
@Transactional
public class KafkaLoginMessageReceiver {

    @KafkaListener(topics = KafkaConstants.KAFKA_ADMIN_LOGIN_STATUS_TOPIC, groupId = KafkaConstants.KAFKA_GROUP_ID)
    public void receiveComment(String message, Acknowledgment acknowledgment) {


        try {
            log.info("收到消息: " + message);

            String[] strings = message.split(":");
            boolean flag = Boolean.parseBoolean(strings[1]);
            String key = strings[0];

            if (!flag) {
                RedisUtils.deleteTokenKey(key);
                // 手动提交偏移量
                if (RedisUtils.getJwtTokenFromRedis(key) == null) {
                    log.info("ID为: " + key.split("-")[1] + " 退出登录成功!");
                    acknowledgment.acknowledge();
                }
            } else {
                if (RedisUtils.getJwtTokenFromRedis(key) != null) {
                    log.info("ID为: " + key.split("-")[1] + " 的用户, 登录成功!");
                    acknowledgment.acknowledge();
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage());
            throw new CustomException(e.getMessage(), StatusCodeEnum.FAILED.getCode());
        }

    }
}
